package com.leilei.source.collection;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;


/**
 * @author lei
 * @version 1.0
 * @date 2021/3/8 21:14
 * @desc flink 基于集合获取数据源
 *
 */
public class FlinkSourceByCollection {
    public static void main(String[] args) {
        try {
            //准备环境 env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置自动识别数据流类型（有界（流）/无界（批））
            env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
            //准备数据源 source
            DataStream<String> fromCollectionSource = env.fromCollection(Arrays.asList("java", "php", "scala", "c", "c++", "python", "java", "c#", "go"));
            DataStream<String> fromElementsSource = env.fromElements("篮球", "排球", "足球", "乒乓球", "保龄球");
            DataStream<Long> fromSequenceSource = env.fromSequence(1L, 10L);
            //逻辑处理

            //数据收集 sink
            fromCollectionSource.print("编程语言>>");

            fromElementsSource.print("球球大作战>>");

            SingleOutputStreamOperator<Tuple2<Long, Long>> map = fromSequenceSource
                    .map(t -> Tuple2.of(t, 1L))
                    .returns(Types.TUPLE(Types.LONG,Types.LONG));
            map.keyBy(t->t.f0).sum(1).print("数字>>");
            //执行 execute
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
